package com.atguigu.ting.realtime.app.dwd.db;

import com.atguigu.ting.realtime.app.BaseSqlApp;
import com.atguigu.ting.realtime.common.Constant;
import com.atguigu.ting.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/4/12 08:33
 */
public class DwdTradeRecharge extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeRecharge().init(
            3006,
            2,
            "DwdTradeRecharge"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db
        readOdsDb(tEnv, "DwdTradeRecharge");
        // 2. 读取字典表
        readBaseDic(tEnv);
        // 3. 过滤出用户充值信息数据
        Table rechargeInfo = tEnv.sqlQuery(
            "select " +
                "`data`['id'] id, " +
                "`data`['user_id'] user_id, " +
                "`data`['trade_no'] trade_no, " +
                "cast(`data`['recharge_amount'] as decimal(16,2)) recharge_amount, " +
                "`data`['payment_type'] payment_type, " +
                "`data`['update_time'] recharge_time," +
                "pt," +
                "ts " +
                "from ods_db " +
                "where `database`='ting' " +
                "and `table`='recharge_info' " +
                "and `type`='update' " +
                "and `old`['recharge_status'] is not null " +
                "and `data`['recharge_status']='1002' ");
    
        tEnv.createTemporaryView("recharge_info", rechargeInfo);
        
        // 4. 两张表 join
        Table result = tEnv.sqlQuery("select " +
                                        "ri.id, " +
                                        "ri.user_id, " +
                                        "ri.trade_no, " +
                                        "ri.payment_type, " +
                                        "dic.dic_name payment_type_name, " +
                                        "ri.recharge_time, " +
                                        "ri.recharge_amount, " +
                                        "ri.ts " +
                                        "from recharge_info ri " +
                                        "join base_dic for system_time as of ri.pt as dic " +
                                        "on ri.payment_type=dic.dic_code ");
        // 5. 写出到 kafka
        tEnv.executeSql("create table dwd_trade_recharge(" +
                            "id string, " +
                            "user_id string, " +
                            "trade_no string, " +
                            "payment_type string, " +
                            "payment_type_name string, " +
                            "recharge_time string, " +
                            "recharge_amount decimal(16,2), " +
                            "ts bigint " +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_RECHARGE));
        result.executeInsert("dwd_trade_recharge");
    }
}
